package cn.tank.data.sync.kafka.common;

import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/************************************************
 * @Description:kafka管理器，需要优化
 *              最好是单例模式
 * @Author: tancyong
 * @Version: 1.0
 ************************************************/
public class KafkaManager {

    private KafkaProperties kafkaProperties;

    private Producer<String, String> producer;

    public KafkaManager(KafkaProperties kafkaProperties) {

        this.kafkaProperties = kafkaProperties;
        this.initProducer();
    }

    private Properties createProducterProperties() {
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaProperties.getKafkaServer());
        props.put("batch.size", kafkaProperties.getBatchSize());
        props.put("acks", kafkaProperties.getAcks());
        props.put("retries", kafkaProperties.getRetries());
        props.put("request.timeout.ms", kafkaProperties.getReqTimeoutMs());
        props.put("retries.backoff.ms", kafkaProperties.getRetriesBackoffMs());
        props.put("linger.ms", kafkaProperties.getLingerMs());
        props.put("buffer.memory", kafkaProperties.getBufferMemory());
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());

        return props;
    }

    private Properties createConsumerProperties(String groupId) {
        Properties props = new Properties();

        props.put("bootstrap.servers", kafkaProperties.getKafkaServer());
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("group.id", StrUtil.isEmpty(groupId)?UUID.randomUUID().toString():groupId);

        return props;
    }

    private void initProducer() {
        try {
            producer = new KafkaProducer<>(createProducterProperties());
        } catch (Exception e) {
            producer = null;
        }
    }


    public Producer<String, String> getProducer() {
        if (producer == null) {
            throw new RuntimeException("Kafka Producer is null");
        }
        return producer;
    }

    public Consumer<String,String> getConsumer(String groupId){

        Properties properties = createConsumerProperties(groupId);
        return new KafkaConsumer<>(properties);
    }

}
